package net.lab1024.sa.base.utils;

import lombok.extern.slf4j.Slf4j;
import net.lab1024.sa.base.module.support.syslog.domain.entity.SyslogConfigEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.net.*;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Syslog 日志发送工具类
 * 支持 UDP、TCP(TLS) 和 Kafka 协议
 *
 * @Author 陈玉强
 * @Date 2025-10-20
 */
@Slf4j
@Component
public class SyslogSender {

    private static final ConcurrentHashMap<String, Object> senderCache = new ConcurrentHashMap<>();

    /**
     * 协议类型常量
     */
    public static final String PROTOCOL_UDP = "UDP";
    public static final String PROTOCOL_TCP = "TCP";
    public static final String PROTOCOL_KAFKA = "KAFKA";

    /**
     * 发送日志
     *
     * @param config Syslog配置
     * @param message 日志消息
     * @return 是否发送成功
     */
    public boolean send(SyslogConfigEntity config, String message) {
        message = JasyptUtil.encrypt(message);
        if (config == null || !Boolean.TRUE.equals(config.getForwardEnabled())) {
            return false;
        }

        try {
            String protocol = config.getProtocolType();
            if (StringUtils.isBlank(protocol)) {
                log.error("Protocol type is empty");
                return false;
            }

            switch (protocol.toUpperCase()) {
                case PROTOCOL_UDP:
                    return sendUdp(config, message);
                case PROTOCOL_TCP:
                    return sendTcp(config, message);
                case PROTOCOL_KAFKA:
                    return sendKafka(config, message);
                default:
                    log.error("Unsupported protocol type: {}", protocol);
                    return false;
            }
        } catch (Exception e) {
            log.error("Failed to send syslog message", e);
            return false;
        }
    }

    /**
     * 发送Kafka日志
     */
    private boolean sendKafka(SyslogConfigEntity config, String message) {
        String cacheKey = "KAFKA_" + config.getSyslogServerIp() + "_" + config.getSyslogServerPort();
        KafkaProducer<String, String> producer = (KafkaProducer<String, String>) senderCache.computeIfAbsent(cacheKey, k -> {
            Properties props = new Properties();
            // 设置Kafka服务器地址和端口
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
                     config.getSyslogServerIp() + ":" + config.getSyslogServerPort());
            
            // 设置序列化器
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            
            // 如果启用了TLS
            if (Boolean.TRUE.equals(config.getTlsEnabled())) {
                props.put("security.protocol", "SSL");
                // 如果需要其他SSL配置，可以在这里添加
            }
            
            // 如果配置了用户名和密码，设置SASL认证
            if (StringUtils.isNotBlank(config.getPassword())) {
                props.put("security.protocol", "SASL_PLAINTEXT");
                props.put("sasl.mechanism", "PLAIN");
                props.put("sasl.jaas.config", config.getPassword());
            }

            // 设置生产者配置
            props.put(ProducerConfig.ACKS_CONFIG, "1"); // 至少等待leader副本确认
            props.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
            props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000); // 重试间隔
            props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 6000); // 发送超时时间
            
            try {
                return new KafkaProducer<>(props);
            } catch (Exception e) {
                log.error("Failed to create Kafka producer for " + cacheKey, e);
                return null;
            }
        });

        if (producer == null) {
            senderCache.remove(cacheKey);
            return false;
        }

        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(
                config.getTopic(),
                message
            );
            // 同步发送消息
            producer.send(record).get();
            return true;
        } catch (Exception e) {
            log.error("Failed to send message to Kafka", e);
            senderCache.remove(cacheKey);
            return false;
        }
    }

    /**
     * 发送UDP日志
     */
    private boolean sendUdp(SyslogConfigEntity config, String message) throws IOException {
        String cacheKey = "UDP_" + config.getSyslogServerIp() + "_" + config.getSyslogServerPort();
        DatagramSocket socket = (DatagramSocket) senderCache.computeIfAbsent(cacheKey, k -> {
            try {
                return new DatagramSocket();
            } catch (SocketException e) {
                log.error("Failed to create UDP socket", e);
                return null;
            }
        });

        if (socket == null) {
            return false;
        }

        byte[] messageBytes = message.getBytes();
        DatagramPacket packet = new DatagramPacket(
            messageBytes,
            messageBytes.length,
            InetAddress.getByName(config.getSyslogServerIp()),
            Integer.parseInt(config.getSyslogServerPort())
        );

        socket.send(packet);
        return true;
    }

    /**
     * 发送TCP日志
     */
    private boolean sendTcp(SyslogConfigEntity config, String message) throws IOException {
        String cacheKey = "TCP_" + config.getSyslogServerIp() + "_" + config.getSyslogServerPort();
        Socket socket;

        if (Boolean.TRUE.equals(config.getTlsEnabled())) {
            // TLS连接
            SSLSocketFactory sslSocketFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
            socket = (Socket) senderCache.computeIfAbsent(cacheKey, k -> {
                try {
                    SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(
                        config.getSyslogServerIp(),
                        Integer.parseInt(config.getSyslogServerPort())
                    );
                    sslSocket.startHandshake();
                    return sslSocket;
                } catch (IOException e) {
                    log.error("Failed to create SSL socket", e);
                    return null;
                }
            });
        } else {
            // 普通TCP连接
            socket = (Socket) senderCache.computeIfAbsent(cacheKey, k -> {
                try {
                    return new Socket(config.getSyslogServerIp(), Integer.parseInt(config.getSyslogServerPort()));
                } catch (IOException e) {
                    log.error("Failed to create TCP socket", e);
                    return null;
                }
            });
        }

        if (socket == null || socket.isClosed()) {
            senderCache.remove(cacheKey);
            return false;
        }

        socket.getOutputStream().write((message + "\n").getBytes());
        socket.getOutputStream().flush();
        return true;
    }

    /**
     * 关闭所有连接
     */
    @PreDestroy
    public void closeAll() {
        senderCache.forEach((key, sender) -> {
            try {
                if (sender instanceof Socket) {
                    ((Socket) sender).close();
                } else if (sender instanceof DatagramSocket) {
                    ((DatagramSocket) sender).close();
                } else if (sender instanceof KafkaProducer) {
                    ((KafkaProducer<?, ?>) sender).close();
                }
            } catch (Exception e) {
                log.error("Failed to close sender: " + key, e);
            }
        });
        senderCache.clear();
    }
}